package cn.ymatrix.hadoopEcology.readParquetTOYMatrix;

import cn.ymatrix.utils.Config;
import com.qcloud.cos.COSClient;
import com.qcloud.cos.ClientConfig;
import com.qcloud.cos.auth.BasicCOSCredentials;
import com.qcloud.cos.auth.COSCredentials;
import com.qcloud.cos.exception.CosClientException;
import com.qcloud.cos.exception.CosServiceException;
import com.qcloud.cos.http.HttpProtocol;
import com.qcloud.cos.model.*;
import com.qcloud.cos.region.Region;
import com.qcloud.cos.transfer.Download;
import com.qcloud.cos.transfer.TransferManager;
import org.apache.commons.lang3.time.StopWatch;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.parquet.example.data.Group;
import org.apache.parquet.example.data.simple.SimpleGroup;
import org.apache.parquet.hadoop.Footer;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.ParquetReader;
import org.apache.parquet.hadoop.example.GroupReadSupport;
import org.apache.parquet.hadoop.metadata.BlockMetaData;

import java.io.*;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class AnalysisForOssParquetTOYMatrix {

  public static void main(String[] args) throws IOException, SQLException {

    String regionStr = Config.getRegionStr();
    String localFileName = Config.getCsvFilePath();
    COSCredentials cred = new BasicCOSCredentials(Config.getSecretId(), Config.getSecretKey());
    Region region = new Region(regionStr);
    ClientConfig clientConfig = new ClientConfig(region);
    clientConfig.setHttpProtocol(HttpProtocol.https);
    COSClient cosClient = new COSClient(cred, clientConfig);
    List<String> fileNameList = readFileList(cosClient);

    for (int i = 0; i < fileNameList.size(); i++) {
      String saveFileName = localFileName + i + ".parquet";
      StopWatch startWatch = StopWatch.createStarted();
      String fileKey = fileNameList.get(i);
      GetObjectRequest getObjectRequest = new GetObjectRequest(Config.getBucketName(), fileKey);
      COSObject cosObject = cosClient.getObject(getObjectRequest);
      cosObject.getObjectContent();
      // 过滤符合条件的数据
      if (fileKey.contains(".")) {
        System.out.println("开始下载 " + fileKey + "文件....");
        downloadFile(cosClient, fileKey, saveFileName);
        System.out.println(
            "下载" + fileKey + "文件完成,耗时" + startWatch.getTime(TimeUnit.MILLISECONDS) + " 毫秒");
        // 解析数据并进行入库
        analysisParquetTOYMatrix(saveFileName);
      }
    }

    // 退出程序
    // cosClient.shutdown();
    // System.exit(0);
  }

  public static void analysisParquetTOYMatrix(String filepath) throws IOException, SQLException {
    ParquetReader.Builder<Group> reader =
        ParquetReader.builder(new GroupReadSupport(), new Path(filepath));
    ParquetReader<Group> parquetReader = reader.build();
    SimpleGroup record;

    long totalNumber = getParquetRowCount(filepath);
    // JDBC connent
    Connection connection = Config.getDateSource().getConnection();
    PreparedStatement preparedStatement = null;
    connection.setAutoCommit(false);

    int batchSize = Config.getBatchSize();
    long indexNum = 0;

    long startTime = System.currentTimeMillis();
    System.out.println("---- 开始插入数据 ----");
    while ((record = (SimpleGroup) parquetReader.read()) != null) {
      indexNum++;
      String[] lineDataSplit = record.toString().split("\n");
      StringBuilder sqlInsert = new StringBuilder("INSERT INTO " + Config.getTableName() + "(");
      StringBuilder sqlStatement = new StringBuilder();
      for (int i = 0; i < lineDataSplit.length; i++) {
        String key = lineDataSplit[i].split(":")[0];
        sqlInsert.append(key + ",");
        sqlStatement.append("?" + ",");
      }
      sqlInsert.deleteCharAt(sqlInsert.length() - 1).toString();
      sqlInsert.append(") VALUES (");

      String insertSQL =
          sqlInsert
              .append(sqlStatement.deleteCharAt(sqlStatement.length() - 1).append(")"))
              .toString();

      preparedStatement = connection.prepareStatement(insertSQL);

      for (int i = 0; i < lineDataSplit.length; i++) {
        String value = lineDataSplit[i].split(":")[1].replaceAll("\\s*", "");
        preparedStatement.setObject(i + 1, value);
      }
      preparedStatement.addBatch();
      preparedStatement.executeBatch();
      if (indexNum % batchSize == 0) {
        long analysisStartTime = System.currentTimeMillis() - startTime;
        long insertStartTime = System.currentTimeMillis();

        connection.commit();
        preparedStatement.clearBatch();

        System.out.println(
            "成功插入第 "
                + indexNum
                + " 条数据,解释耗时"
                + analysisStartTime
                + " ms,入库耗时"
                + (System.currentTimeMillis() - insertStartTime)
                + "ms");
      }
    }

    // 插入剩下的部分
    preparedStatement.executeBatch();
    preparedStatement.clearBatch();
    connection.commit();
    long spendTime = System.currentTimeMillis() - startTime;
    System.out.println("成功插入 " + totalNumber + " 条数据,共耗时: " + spendTime + "毫秒");

    preparedStatement.close();
    connection.close();
  }

  public static void listPath(COSClient cosClient) {
    List<Bucket> buckets = cosClient.listBuckets();
    for (Bucket bucketElement : buckets) {
      String bucketName = bucketElement.getName();
      String bucketLocation = bucketElement.getLocation();
      System.out.println("bucketName:" + bucketName + "   bucketLocation :" + bucketLocation);
    }
  }

  public static void downloadFile(COSClient cosClient, String key, String localFileName) {
    TransferManager transferManager = createTransferManager(cosClient);
    File downloadFile = new File(localFileName);
    GetObjectRequest getObjectRequest = new GetObjectRequest(Config.getBucketName(), key);
    try {
      Download download = transferManager.download(getObjectRequest, downloadFile);
      download.waitForCompletion();
    } catch (CosServiceException e) {
      e.printStackTrace();
    } catch (CosClientException e) {
      e.printStackTrace();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }

  public static TransferManager createTransferManager(COSClient cosClient) {
    ExecutorService threadPool = Executors.newFixedThreadPool(Config.getFixedThreadPool());
    TransferManager transferManager = new TransferManager(cosClient, threadPool);
    return transferManager;
  }

  public static List<String> readFileList(COSClient cosClient) {
    List<String> fileNameList = new ArrayList<>();
    ListObjectsRequest listObjectsRequest = new ListObjectsRequest();
    listObjectsRequest.setBucketName(Config.getBucketName());
    listObjectsRequest.setPrefix(Config.getPrefix() + "/");
    listObjectsRequest.setDelimiter("/");
    listObjectsRequest.setMaxKeys(1000);
    ObjectListing objectListing = null;
    do {
      try {
        objectListing = cosClient.listObjects(listObjectsRequest);
      } catch (CosServiceException e) {
        e.printStackTrace();
      } catch (CosClientException e) {
        e.printStackTrace();
      }
      List<COSObjectSummary> cosObjectSummaries = objectListing.getObjectSummaries();
      for (COSObjectSummary cosObjectSummary : cosObjectSummaries) {
        String key = cosObjectSummary.getKey();
        fileNameList.add(key);
        //  System.out.println("key:" + key);
      }
    } while (objectListing.isTruncated());
    return fileNameList;
  }

  public static long getParquetRowCount(String filepath) {
    Path inputPath = new Path(filepath);
    Configuration conf = new Configuration();
    long rowCount = 0;
    try {
      FileStatus[] inputFileStatuses = inputPath.getFileSystem(conf).globStatus(inputPath);
      for (FileStatus fs : inputFileStatuses) {
        for (Footer f : ParquetFileReader.readFooters(conf, fs, false)) {
          for (BlockMetaData b : f.getParquetMetadata().getBlocks()) {
            rowCount = b.getRowCount();
          }
        }
      }
    } catch (IOException e) {
      e.printStackTrace();
    }
    return rowCount;
  }

  public static String readLineNumber(String filePath, int lineNumber) throws IOException {
    String appointedLine = "";
    FileReader in = null;
    LineNumberReader reader = null;
    try {
      in = new FileReader(filePath);
      reader = new LineNumberReader(in);
      long totalLine = countTextLines(filePath);
      if (lineNumber < 0 || lineNumber > totalLine) {
        return null;
      }
      int line = 1;
      reader.setLineNumber(lineNumber);
      long i = reader.getLineNumber();
      String s = "";
      while ((s = reader.readLine()) != null) {
        if (i == line) {
          appointedLine = s;
          break;
        }
        line++;
      }
      return appointedLine;
    } catch (FileNotFoundException e) {
      e.printStackTrace();
    } catch (IOException e) {
      e.printStackTrace();
    } catch (Exception e) {
      e.printStackTrace();
    } finally {
      reader.close();
      in.close();
    }
    return appointedLine;
  }

  public static boolean containRepeatChar(String sourceStr, String targetStr) {
    boolean statusValue = false;
    int index = sourceStr.lastIndexOf(targetStr);
    if (index > 0) {
      statusValue = true;
    }
    return statusValue;
  }

  public static Integer countTextLines(String filePath) {
    Integer lines = 0;
    try {
      File file = new File(filePath);
      if (file.exists()) {
        long fileLength = file.length();
        LineNumberReader lineNumberReader = new LineNumberReader(new FileReader(file));
        lineNumberReader.skip(fileLength);
        lines = lineNumberReader.getLineNumber();
        System.out.println("Total number of lines : " + lines);
        lineNumberReader.close();
      } else {
        System.out.println("File does't exists");
      }
    } catch (IOException e) {
      e.printStackTrace();
    }
    return lines;
  }
}
